package org.hepeng.workx.spring.cloud.netflix.eureka.server;

import io.reactivex.Observable;
import io.reactivex.schedulers.Schedulers;
import org.hepeng.workx.util.concurrent.NamedThreadFactory;
import org.springframework.context.ApplicationEvent;

import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author he peng
 */
public class AsyncEurekaEventConsumerExecutor implements EurekaEventConsumerExecutor {

    private Executor executor;

    public AsyncEurekaEventConsumerExecutor() {
        ThreadFactory threadFactory = new NamedThreadFactory("EurekaEventConsumer" , true);
        this.executor = new ThreadPoolExecutor(2 , 2 , 0L , TimeUnit.MILLISECONDS , new LinkedBlockingQueue<>() , threadFactory);
    }

    @Override
    public void execute(ApplicationEvent event, List<EurekaServerEventConsumer> consumers) {
        Observable<ApplicationEvent> observable = Observable.just(event)
                .subscribeOn(Schedulers.from(this.executor));
        consumers.forEach(
                consumer -> observable.subscribe(
                        data -> consumer.onConsume(data) ,
                        error -> consumer.onError(error),
                        () -> consumer.onComplete()));
    }
}
